#!/usr/bin/env python
# -*- coding: utf-8 -*-

# from tornado.iostream import IOStream
from tcpclient import TCPClient
from tornado import ioloop, stack_context, gen, iostream
import functools

from collections import deque
from dm_py_util import DEBUG_LOG, ERROR_LOG, INFO_LOG
import datetime
import sub_protocol
import socket
import struct
import time


class SubscribeClient(object):

    def __init__(self, path, topic, timeout=1):
        self.topic = topic
        self.path = path
        self.timeout = None
        self.stream = None
        self.connected = False
        self.last_active_time = 0
        self.last_heartbeat = 0
        self.connect_timeout = timeout
        self.buf = deque()
        self.pc = ioloop.PeriodicCallback(functools.partial(
            stack_context.wrap(self.do_keepalive)), 500)

    @gen.coroutine
    def connect(self):
        self.connected = False
        host, port = self.path
        DEBUG_LOG('connect to host', host, 'port', port)
        tcpclient = TCPClient()
        try:
            self.stream = yield tcpclient.connect(host, port, af=socket.AF_INET)
        except Exception as e:
            ERROR_LOG('connect failed, path', self.path, e)
            self.on_close()
        else:
            yield self.on_connected()
            self.stream.set_close_callback(stack_context.wrap(self.on_close))

    @gen.coroutine
    def on_connected(self):
        DEBUG_LOG('on connected', len(self.buf))
        self.stream.set_nodelay(True)
        self.timeout = None
        self.pc.start()
        self.subcribe()
        self.last_heartbeat = self.last_active_time = time.time()
        # data = ''.join(self.buf)
        self.buf.clear()
        # self.send_message(data)
        self.connected = True
        if not self.stream.closed():
            self.read_data()

    def subcribe(self):
        if not self.stream.closed():
            s = sub_protocol.get_only_pub_msg(self.topic)
            f = self.stream.write(s)
            ioloop.IOLoop.current().add_future(f, lambda r: r.result())

    def close(self):
        if self.stream and not self.stream.closed():
            self.stream.close()

    def on_close(self):
        DEBUG_LOG('subclient connection onclose')
        self.pc.stop()
        if self.timeout:
            ioloop.IOLoop.current().remove_timeout(self.timeout)
        self.timeout = ioloop.IOLoop.current().add_timeout(
            datetime.timedelta(milliseconds=100), stack_context.wrap(self.connect))

    def read_data(self):
        try:
            self.stream.read_bytes(6, self.on_header)
        except Exception as e:
            ERROR_LOG('read mesasage header error close connect', e)
            self.close()

    def on_header(self, data):
        try:
            msg_len, head_len = struct.unpack('!1Ih', data)
            # DEBUG_LOG('msg len', msg_len, 'head len', head_len)

            def on_body(data):
                cmd,  = struct.unpack('!1B', data[0])
                if cmd == 1:
                    rlt,  = struct.unpack('!1i', data[5:9])
                    if rlt < 0:
                        INFO_LOG('订阅失败', self.topic)
                    else:
                        INFO_LOG('订阅成功', self.topic)
                elif cmd == 4:
                    self.on_heartbeat()
                    return
                head = data[:head_len]
                DEBUG_LOG('data', data[head_len + 5:])
            self.stream.read_bytes(msg_len, on_body)
            self.read_data()
        except Exception as e:
            ERROR_LOG('on mesasage header error close connect', e)
            self.close()

    def on_heartbeat(self):
        self.last_active_time = int(time.time())
        DEBUG_LOG('收到心跳, 更新心跳时间')

    def do_keepalive(self):
        if self.stream and not self.stream.closed():
            now = int(time.time())
            if now - self.last_heartbeat > 5:
                s = sub_protocol.get_heartbeat_msg(self.topic)
                self.stream.write(s, self.write_callback)
                self.last_heartbeat = now
                DEBUG_LOG('发送心跳, 心跳包长度', len(s))
            if (self.last_active_time - self.last_heartbeat) > 2 and not self.stream.closed():
                DEBUG_LOG('心跳超时, 断掉连接')
                self.stream.close()
        else:
            self.on_close()

    def write_callback(self):
        # DEBUG_LOG('write success')
        pass

    def send_message(self, data):
        ms = struct.pack('!I', len(data)) + '#' + \
            struct.pack('%ds' % len(data), data)
        data = sub_protocol.get_pub_msg(self.topic, ms)
        if not self.stream or self.stream.closed() or not self.connected:
            #     if len(self.buf) > 100:
            #         self.buf.clear()
            #     self.buf.append(data)
            #     INFO_LOG(
            #         'send message connection not build push to buffer', len(self.buf))
            return
        # if len(self.buf):
        #     d = ''.join(self.buf)
        #     self.buf.clear()
        #     data = ''.join([d, data])
        try:
            self.stream.write(data, self.write_callback)
        except Exception as e:
            ERROR_LOG('send message error close connect', e)
            self.close()


if __name__ == '__main__':
    from tornado import ioloop, stack_context
    from extend import set_timeout, set_interval
    import functools

    s = SubscribeClient(('192.168.20.128', 6045), 'cluster_1')

    @gen.coroutine
    def connect():
        yield s.connect()

    def interval():
        s.send_message('hahhahahah')
        # pass

    set_timeout(connect, 0)
    # ioloop.IOLoop.current().run_sync(stack_context.wrap(s.connect))
    set_interval(interval, 1)
    ioloop.IOLoop.instance().start()
